package com.sui.bigdata.rtcadmin.testEnv;

import com.feidee.fdcommon.model.StringResponse;
import com.feidee.fdcommon.util.HttpUtil;
import org.apache.http.Consts;
import org.apache.http.client.config.RequestConfig;
import org.apache.http.client.methods.HttpPost;
import org.apache.http.entity.StringEntity;
import org.junit.Test;

/**
 * @author YongChen
 * @date 2020/3/9 15:42
 * @description
 * @email yong_chen@sui.com
 */
public class restart {

    @Test
    public void test1() throws  Exception{
        String url = "http://10.201.7.132:18001/rtc-admin/job?action=restart";

        String sql =
                "CREATE TABLE behaviour_common(\n" +
                        "    ssjid varchar,\n" +
                        "    event_time varchar,\n" +
                        "    title varchar ,\n" +
                        "    etype varchar,\n" +
                        "    sku varchar\n" +
                        " )WITH(\n" +
                        "    type ='kafka',\n" +
                        "    bootstrapServers ='10.201.7.187:9093,10.201.7.188:9093,10.201.7.189:9093,10.201.7.190:9093',\n" +
                        "    offsetReset ='latest',\n" +
                        "    groupid ='fin_behaviour_common_2020/5/6',\n" +
                        "    topic ='lc.financing.behaviour_common',\n" +
                        "    parallelism ='1'\n" +
                        " );\n" +
                        " \n" +
                        " \n" +
                        " CREATE TABLE bdl_finance_zbank_account_account(\n" +
                        "    rowData.ssj_id varchar as ssj_id,\n" +
                        "    rowData.create_time varchar as create_time ,\n" +
                        "    rowData.account_status varchar as account_status\n" +
                        " )WITH(\n" +
                        "    type ='kafka',\n" +
                        "    bootstrapServers ='10.201.7.187:9093,10.201.7.188:9093,10.201.7.189:9093,10.201.7.190:9093',\n" +
                        "    offsetReset ='latest',\n" +
                        "    groupid ='bdl_finance_zbank_account_account_2020/5/6',\n" +
                        "    topic ='rtsync.bdl_finance_zbank_account_account',\n" +
                        "    parallelism ='1'\n" +
                        " );\n" +
                        " \n" +
                        " \n" +
                        " CREATE TABLE bdl_finance_zbank_trade_account_order(\n" +
                        "    rowData.ssj_id varchar as ssj_id,\n" +
                        "    rowData.order_finish_time varchar as order_finish_time ,\n" +
                        "    rowData.order_status varchar as order_status,\n" +
                        "\trowData.trade_type varchar as trade_type,\n" +
                        "\trowData.product_code varchar as product_code\n" +
                        " )WITH(\n" +
                        "    type ='kafka',\n" +
                        "    bootstrapServers ='10.201.7.187:9093,10.201.7.188:9093,10.201.7.189:9093,10.201.7.190:9093',\n" +
                        "    offsetReset ='latest',\n" +
                        "    groupid ='bdl_finance_zbank_trade_account_order_2020/5/6',\n" +
                        "    topic ='rtsync.bdl_finance_zbank_trade_account_order',\n" +
                        "    parallelism ='1'\n" +
                        " );\n" +
                        " \n" +
                        " \n" +
                        "  CREATE TABLE bdl_finance_zbank_trade_order(\n" +
                        "    rowData.ssj_id varchar as ssj_id,\n" +
                        "    rowData.create_time varchar as create_time ,\n" +
                        "    rowData.order_status varchar as order_status,\n" +
                        "\trowData.trade_type varchar as trade_type,\n" +
                        "\trowData.product_code varchar as product_code\n" +
                        " )WITH(\n" +
                        "    type ='kafka',\n" +
                        "    bootstrapServers ='10.201.7.187:9093,10.201.7.188:9093,10.201.7.189:9093,10.201.7.190:9093',\n" +
                        "    offsetReset ='latest',\n" +
                        "    groupid ='bdl_finance_zbank_trade_account_order_2020/5/6',\n" +
                        "    topic ='rtsync.bdl_finance_zbank_trade_order',\n" +
                        "    parallelism ='1'\n" +
                        " );\n" +
                        " \n" +
                        " \n" +
                        "CREATE TABLE sideTable (\n" +
                        "   user_id varchar ,\n" +
                        "   suid varchar ,\n" +
                        "   PRIMARY KEY(user_id),\n" +
                        "   PERIOD FOR SYSTEM_TIME\n" +
                        " ) WITH (\n" +
                        " type='mysql',\n" +
                        " url='jdbc:mysql://tidb1.testfeideedba.com:4000/dt?useUnicode=true&amp;characterEncoding=utf8;useSSL=true',\n" +
                        " userName='DSJ_user',\n" +
                        " password='606cde3507b026H,dHb_77e67H_e',\n" +
                        " tableName='adl_bd_ls_ssj_kn_userid_udid_suid_tidb',\n" +
                        " maxPoolSize = '100' , \n" +
                        " checkOutTimeout = '100000');\n" +
                        " \n" +
                        "CREATE TABLE view_product_page(\n" +
                        "     udidmd5 varchar,\n" +
                        "     sid varchar,\n" +
                        "     hbase_table varchar,\n" +
                        "     userssj_zbank_view_product_page_at_last_time varchar\n" +
                        "  )WITH(\n" +
                        "     type ='kafka',\n" +
                        "     bootstrapServers ='10.201.7.187:9093,10.201.7.188:9093,10.201.7.189:9093,10.201.7.190:9093',\n" +
                        "     topic ='rttag.realtime.tag_total',\n" +
                        "     parallelism ='1'\n" +
                        "  );\n" +
                        "  \n" +
                        "  CREATE TABLE open_account(\n" +
                        "     udidmd5 varchar,\n" +
                        "     sid varchar,\n" +
                        "     hbase_table varchar,\n" +
                        "     userssj_zbank_open_account_at_last_time varchar\n" +
                        "  )WITH(\n" +
                        "     type ='kafka',\n" +
                        "     bootstrapServers ='10.201.7.187:9093,10.201.7.188:9093,10.201.7.189:9093,10.201.7.190:9093',\n" +
                        "     topic ='rttag.realtime.tag_total',\n" +
                        "     parallelism ='1'\n" +
                        "  );\n" +
                        "  \n" +
                        "  \n" +
                        "  CREATE TABLE recharge(\n" +
                        "     udidmd5 varchar,\n" +
                        "     sid varchar,\n" +
                        "     hbase_table varchar,\n" +
                        "     userssj_zbank_recharge_at_last_time varchar\n" +
                        "  )WITH(\n" +
                        "     type ='kafka',\n" +
                        "     bootstrapServers ='10.201.7.187:9093,10.201.7.188:9093,10.201.7.189:9093,10.201.7.190:9093',\n" +
                        "     topic ='rttag.realtime.tag_total',\n" +
                        "     parallelism ='1'\n" +
                        "  );\n" +
                        "  \n" +
                        "  \n" +
                        "  CREATE TABLE deposit(\n" +
                        "     udidmd5 varchar,\n" +
                        "     sid varchar,\n" +
                        "     hbase_table varchar,\n" +
                        "     userssj_zbank_deposit_at_last_time varchar\n" +
                        "  )WITH(\n" +
                        "     type ='kafka',\n" +
                        "     bootstrapServers ='10.201.7.187:9093,10.201.7.188:9093,10.201.7.189:9093,10.201.7.190:9093',\n" +
                        "     topic ='rttag.realtime.tag_total',\n" +
                        "     parallelism ='1'\n" +
                        "  ); \n" +
                        " \n" +
                        "insert into  view_product_page\n" +
                        "select md5(trim(t2.suid)) as udidmd5,\n" +
                        "      trim(t2.suid) as sid ,\n" +
                        "\t  'ssj_user_tagging' as hbase_table,\n" +
                        "\t  t1.event_time as userssj_zbank_view_product_page_at_last_time \n" +
                        "from (select ssjid as ssj_id,max(substring(event_time,1,19)) as event_time\n" +
                        "     from behaviour_common \n" +
                        "     where title like '产品详情页%' and  LOWER(etype) = 'view' and sku in ('bank_zb','zbank') and NotBlank(ssjid)\n" +
                        "     group by ssjid) t1 \n" +
                        "join sideTable t2 on t1.ssj_id = t2.user_id;\n" +
                        "\n" +
                        "\n" +
                        "insert into  open_account\n" +
                        "select md5(trim(n2.suid)) as udidmd5,\n" +
                        "       trim(n2.suid) as sid,\n" +
                        "\t   'ssj_user_tagging' as hbase_table,\n" +
                        "       n1.create_time as  userssj_zbank_open_account_at_last_time\n" +
                        "from (select ssj_id,max(create_time) as create_time\n" +
                        "      from bdl_finance_zbank_account_account\n" +
                        "      where account_status = '1' and  NotBlank(ssj_id)\n" +
                        "      group by ssj_id) n1 \n" +
                        "join sideTable n2 on n1.ssj_id = n2.user_id;\n" +
                        "\n" +
                        "\n" +
                        "insert into recharge\n" +
                        "select md5(trim(m2.suid)) as udidmd5,\n" +
                        "     trim(m2.suid) as sid,\n" +
                        "\t 'ssj_user_tagging' as hbase_table,\n" +
                        "\t m1.order_finish_time  as  userssj_zbank_recharge_at_last_time \n" +
                        "from (select ssj_id,max(order_finish_time) as order_finish_time\n" +
                        "     from  bdl_finance_zbank_trade_account_order\n" +
                        "     where order_status = '2' and  trade_type = '1' and  lower(product_code)='zbank' and NotBlank(ssj_id)\n" +
                        "     group by ssj_id) m1 \n" +
                        "join sideTable m2 on m1.ssj_id = m2.user_id;\n" +
                        "\n" +
                        "\n" +
                        "insert into deposit\n" +
                        "select \n" +
                        "     md5(trim(k2.suid)) as udidmd5,\n" +
                        "     trim(k2.suid) as sid,\n" +
                        "\t 'ssj_user_tagging' as hbase_table,\n" +
                        "     k1.create_time as userssj_zbank_deposit_at_last_time\n" +
                        "from(select ssj_id,max(create_time) as create_time\n" +
                        "     from  bdl_finance_zbank_trade_order\n" +
                        "     where order_status = '2' and  trade_type = '1' and  lower(product_code)='zbank' and NotBlank(ssj_id)\n" +
                        "     group by ssj_id) k1 \n" +
                        "join sideTable k2 on k1.ssj_id = k2.user_id;\n" ;



//        String data = "{\"jobName\":\"credit_aduit_tag\",\"extendParam\":{\"time.characteristic\":\"EventTime\"}}";

        String data = "{\"jobName\":\"vis_rt_userssj_2tag_20200624\",\"extendParam\":{\"sql.checkpoint.interval\":\"300000\"},\"savepoint\":\"true\"}";



//        String data = "{\"execSql\":\"" + sql +
//                "\",\"jobName\":\"terminal_flink_kudu\"," +
//                "\"savepoint\":\"true\"}";

        System.out.println(data);
        StringEntity stringEntity = new StringEntity(data, Consts.UTF_8);
        stringEntity.setContentEncoding("UTF-8");
        stringEntity.setContentType("application/json");

        HttpPost httpPost=new HttpPost(url);

        RequestConfig requestConfig = RequestConfig.custom().setConnectTimeout(500000).build();
        httpPost.setEntity(stringEntity);
        httpPost.setConfig(requestConfig);
        StringResponse response = HttpUtil.post(httpPost);
        System.out.println(">>>>> "+response.getResponseBody());
    }
}
